"""
Case Type   : Logical_Replication
Case Name   : 使用wal2json插件开启add-tables选项进行逻辑解码
Create At   : 2024/10/21
Owner       : xiaqi14
Description :
    1.修改参数
    2.配置初始用户白名单
    3.安装wal2json插件
    4.创建带表
    5.创建逻辑复制槽，指定插件wal2json
    6.主机和备机分别开启逻辑复制槽，开启add-tables选项
    7.执行DML操作
    8.停止逻辑复制槽
    9.删除逻辑复制槽
    10.资源清理
    11.恢复参数，恢复pg_hba.conf文件
Expect      :
    1.修改参数成功
    2.配置成功
    3.安装成功
    4.创建成功
    5.创建成功
    6.开启成功
    7.执行DML操作成功，解码成功
    8.停止成功
    9.删除成功
    10.清理成功
    11.恢复成功
History     :
"""
import os
import unittest
from yat.test import Node
from yat.test import macro
from testcase.utils.Logger import Logger
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.ComThread import ComThread

primary_sh = CommonSH('PrimaryDbUser')


@unittest.skipIf(1 == primary_sh.get_node_num(), '单机环境不执行')
class LogicReplication(unittest.TestCase):
    def setUp(self):
        self.logger = Logger()
        self.logger.info(f"-----{os.path.basename(__file__)} start-----")
        self.primary_node = Node('PrimaryDbUser')
        self.primary_root = Node('PrimaryRoot')
        self.standby_node = Node('Standby1DbUser')
        self.standby_root = Node('Standby1Root')
        self.sh_standby = CommonSH('Standby1DbUser')
        self.constant = Constant()
        self.common = Common()
        self.table_name_1 = 't_logical_replication_0083_1'
        self.table_name_2 = 't_logical_replication_0083_2'
        self.slot_name_1 = 'slot_logical_replication_0083_1'
        self.slot_name_2 = 'slot_logical_replication_0083_2'
        self.logfile_1 = os.path.join(macro.DB_BACKUP_PATH,
                                      'logfile_logical_replication_0083_1.log')
        self.logfile_2 = os.path.join(macro.DB_BACKUP_PATH,
                                      'logfile_logical_replication_0083_2.log')
        gauss_home = self.primary_node.sh(f'source {macro.DB_ENV_PATH};'
                                          f'ls -d $GAUSSHOME').result()
        self.wal2json_dir = os.path.join(gauss_home, 'lib', 'postgresql')
        self.wal2json_path = os.path.join(self.wal2json_dir, 'wal2json.so')
        self.config_slot_log = 'enable_slot_log'
        self.slot_log_default = self.common.show_param(self.config_slot_log)
        self.config_wal_level = 'wal_level'
        self.wal_level_default = self.common.show_param(self.config_wal_level)

    def test_logical_replication(self):
        text = "-----step1:修改参数;expect:修改参数成功-----"
        self.logger.info(text)
        if 'on' != self.slot_log_default:
            guc_result = primary_sh.execute_gsguc(
                'reload', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_slot_log}=on')
            self.assertTrue(guc_result, '执行失败' + text)
        show_msg = self.common.show_param(self.config_slot_log)
        self.assertEquals(show_msg, 'on', '执行失败' + text)

        if 'logical' != self.wal_level_default:
            guc_result = primary_sh.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_wal_level}=logical')
            self.assertTrue(guc_result, '执行失败' + text)
            restart_msg = primary_sh.restart_db_cluster()
            self.logger.info(restart_msg)
            self.assertTrue(restart_msg, '重启失败')
        show_msg = self.common.show_param(self.config_wal_level)
        self.assertEquals(show_msg, 'logical', '执行失败' + text)

        text = "-----step2:配置初始用户白名单;expect:配置成功-----"
        self.logger.info(text)
        param_1 = f'local replication {self.primary_node.ssh_user}  trust'
        param_2 = f'host  replication {self.primary_node.ssh_user} ' \
                  f'127.0.0.1/32 trust'
        param_3 = f'host  replication {self.primary_node.ssh_user} ' \
                  f'::1/128 trust'
        result_1 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_1)
        result_2 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_2)
        result_3 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_3)
        self.assertTrue(result_1 and result_2 and result_3, '执行失败' + text)

        text = "-----step3:安装wal2json插件;expect:安装成功-----"
        self.logger.info(text)
        self.logger.info('-----获取数据库用户的用户组-----')
        user_groups = f'groups {self.primary_node.ssh_user}'
        self.logger.info(user_groups)
        groups_msg = self.primary_root.sh(user_groups).result()
        self.logger.info(groups_msg)
        self.assertIn(self.primary_node.ssh_user, groups_msg,
                      '执行失败:' + text)
        group = groups_msg.split()[-1]
        self.logger.info('-----主机安装wal2json插件-----')
        self.common.scp_file(self.primary_node, 'wal2json.so',
                             self.wal2json_dir)
        chown_cmd = f'chown {self.primary_node.ssh_user}:{group} ' \
                    f'{self.wal2json_path}'
        self.logger.info(chown_cmd)
        chown_msg = self.primary_root.sh(chown_cmd).result()
        self.logger.info(chown_msg)
        self.assertEquals('', chown_msg, '执行失败:' + text)
        self.logger.info('-----备机安装wal2json插件-----')
        self.common.scp_file(self.standby_node, 'wal2json.so',
                             self.wal2json_dir)
        chown_msg = self.standby_root.sh(chown_cmd).result()
        self.logger.info(chown_msg)
        self.assertEquals('', chown_msg, '执行失败:' + text)

        text = "-----step4:创建表;expect:创建成功-----"
        self.logger.info(text)
        create_cmd = f'''drop table if exists {self.table_name_1},{self.table_name_1};
            create table {self.table_name_1}(
            col_integer integer ,col_date date,col_varchar varchar(20));
            create table {self.table_name_2}(c_int  int);'''
        create_msg = primary_sh.execut_db_sql(create_cmd, dbname='postgres')
        self.logger.info(create_msg)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS,
                      create_msg, '执行失败:' + text)
        self.assertEquals(2, create_msg.count(self.constant.CREATE_TABLE_SUCCESS),
                      '执行失败:' + text)

        text = "-----step5:创建逻辑复制槽，指定插件wal2json;" \
               "expect:创建成功-----"
        self.logger.info(text)
        slot_names = [self.slot_name_1, self.slot_name_2]
        for slot_name in slot_names:
            create_slot_cmd = f"source {macro.DB_ENV_PATH};" \
                              f"pg_recvlogical -d postgres -S {slot_name} " \
                              f"-p {self.primary_node.db_port} --create -P wal2json -s 2 "
            self.logger.info(create_slot_cmd)
            create_slot_msg = self.primary_node.sh(create_slot_cmd).result()
            self.logger.info(create_slot_msg)
            self.assertEquals('', create_slot_msg, '执行失败:' + text)
        text = "-----验证创建逻辑复制槽成功;expect:创建成功-----"
        self.logger.info(text)
        sql_cmd = f"select * from pg_get_replication_slots() " \
                  f"where slot_name like '%slot_logical_replication_0083%';"
        sql_msg = primary_sh.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn(self.slot_name_1 and self.slot_name_2,
                      sql_msg, '执行失败:' + text)

        text = "-----step6:主机和备机分别开启逻辑复制槽，使用wal2json插件" \
               "开启add-tables选项;expect:开启成功-----"
        self.logger.info(text)
        thread_1 = ComThread(primary_sh.exec_pg_recvlogical,
                             args=('--start', 'postgres', self.slot_name_1,
                                   self.primary_node.ssh_user,
                                   f'-f {self.logfile_1} -P wal2json '
                                   f'-o add-tables=public.{self.table_name_1} -s 2',))
        thread_1.setDaemon(True)
        thread_1.start()
        thread_1.join(10)
        self.logger.info(thread_1.get_result())
        text = "-----验证主机开启逻辑复制槽成功;expect:开启成功-----"
        self.logger.info(text)
        sql_cmd = f"select active from pg_get_replication_slots() " \
                  f"where slot_name like '{self.slot_name_1}%';;"
        sql_msg = primary_sh.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertEquals('t', sql_msg.splitlines()[-2].strip(),
                          '执行失败:' + text)
        thread_2 = ComThread(self.sh_standby.exec_pg_recvlogical,
                             args=('--start', 'postgres', self.slot_name_2,
                                   self.primary_node.ssh_user,
                                   f'-f {self.logfile_2} -P wal2json '
                                   f'-o add-tables=public.{self.table_name_1} -s 2',))
        thread_2.setDaemon(True)
        thread_2.start()
        thread_2.join(10)
        self.logger.info(thread_2.get_result())
        text = "-----验证备机开启逻辑复制槽成功;expect:开启成功-----"
        self.logger.info(text)
        sql_cmd = f"select active from pg_get_replication_slots() " \
                  f"where slot_name like '{self.slot_name_2}%';;"
        sql_msg = self.sh_standby.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertEquals('t', sql_msg.splitlines()[-2].strip(),
                          '执行失败:' + text)

        text = "-----step7:执行DML操作;expect:执行DML操作成功-----"
        self.logger.info(text)
        sql_cmd = f'''insert into {self.table_name_1} 
            values(1,date '2022-05-22','测试');
            update {self.table_name_1} set col_integer=2,col_varchar='开发';
            delete from {self.table_name_1};
            insert into {self.table_name_2} values(1);'''
        sql_msg = primary_sh.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn(
                      self.constant.UPDATE_SUCCESS_MSG and
                      self.constant.DELETE_SUCCESS_MSG,
                      sql_msg, "执行失败" + text)

    def tearDown(self):
        text = "-----step8:停止逻辑复制槽;expect:停止成功-----"
        self.logger.info(text)
        stop_cmd = "ps -ef | grep  pg_recvlogical | grep -v grep | " \
                   "awk '{{print $2}}' | xargs kill -9"
        self.logger.info(stop_cmd)
        primary_result = self.primary_node.sh(stop_cmd).result()
        self.logger.info(primary_result)
        standby_result = self.standby_node.sh(stop_cmd).result()
        self.logger.info(standby_result)
        sql_cmd = f"select active from pg_get_replication_slots() " \
                  f"where slot_name like '%{self.slot_name_1}%';"
        primary_active_msg = primary_sh.execut_db_sql(sql_cmd,
                                                      dbname='postgres')
        self.logger.info(primary_active_msg)
        sql_cmd = f"select active from pg_get_replication_slots() " \
                  f"where slot_name like '%{self.slot_name_2}%';"
        standby_active_msg = self.sh_standby.execut_db_sql(sql_cmd,
                                                           dbname='postgres')
        self.logger.info(standby_active_msg)

        text_1 = "-----step9:删除逻辑复制槽;expect:删除成功-----"
        self.logger.info(text_1)
        slot_names = [self.slot_name_1, self.slot_name_2]
        for slot_name in slot_names:
            drop_slot_msg = primary_sh.exec_pg_recvlogical(
                '--drop', 'postgres', slot_name,
                self.primary_node.ssh_user, '-s 2')
            self.logger.info(drop_slot_msg)
        sql_cmd = f"select * from pg_get_replication_slots() " \
                  f"where slot_name like '%slot_logical_replication_0080%';"
        sql_msg = primary_sh.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)

        text_2 = "-----step10:资源清理;expect:清理成功-----"
        self.logger.info(text_2)
        drop_cmd = f'''drop table {self.table_name} cascade;'''
        drop_msg = primary_sh.execut_db_sql(drop_cmd, dbname='postgres')
        self.logger.info(drop_msg)
        del_cmd = f'rm -rf {self.logfile_1} {self.wal2json_path};'
        self.logger.info(del_cmd)
        primary_del_res = self.primary_node.sh(del_cmd).result()
        self.logger.info(primary_del_res)
        del_cmd = f'rm -rf {self.logfile_2} {self.wal2json_path};'
        self.logger.info(del_cmd)
        standby_del_res = self.standby_node.sh(del_cmd).result()
        self.logger.info(standby_del_res)
        file_cmd = f'''if [[ -e {self.logfile_1} && -e {self.wal2json_path} ]];
            then echo "exists"; else echo "not exists"; fi'''
        self.logger.info(file_cmd)
        primary_file_msg = self.primary_node.sh(file_cmd).result()
        self.logger.info(primary_file_msg)
        file_cmd = f'''if [[ -e {self.logfile_2} && -e {self.wal2json_path} ]];
                    then echo "exists"; else echo "not exists"; fi'''
        self.logger.info(file_cmd)
        standby_file_msg = self.standby_node.sh(file_cmd).result()
        self.logger.info(standby_file_msg)

        text_3 = "-----step11:恢复参数，恢复pg_hba.conf文件，" \
                 "expect:恢复成功-----"
        self.logger.info(text_3)
        slot_log_current = self.common.show_param(self.config_slot_log)
        if slot_log_current != self.slot_log_default:
            primary_sh.execute_gsguc(
                'reload', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_slot_log}={self.slot_log_default}')
        show_slot_log = self.common.show_param(self.config_slot_log)

        wal_level_current = self.common.show_param(self.config_wal_level)
        if wal_level_current != self.wal_level_default:
            primary_sh.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_wal_level}={self.wal_level_default}')
            restart_msg = primary_sh.restart_db_cluster()
            self.logger.info(restart_msg)
        show_wal_level = self.common.show_param(self.config_wal_level)

        param_1 = f'local replication {self.primary_node.ssh_user} '
        param_2 = f'host replication {self.primary_node.ssh_user} 127.0.0.1/32 '
        param_3 = f'host  replication {self.primary_node.ssh_user} ::1/128 '
        result_1 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_1)
        result_2 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_2)
        result_3 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_3)

        self.assertEquals('f', primary_active_msg.splitlines()[-2].strip(),
                          '执行失败:' + text)
        self.assertEquals('f', standby_active_msg.splitlines()[-2].strip(),
                          '执行失败:' + text)
        self.assertIn("(0 rows)", sql_msg, '执行失败:' + text_1)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS,
                      drop_msg, '执行失败' + text_2)
        self.assertEqual("not exists", primary_file_msg, '执行失败' + text_2)
        self.assertEqual("not exists", standby_file_msg, '执行失败' + text_2)
        self.assertEquals(show_slot_log, self.slot_log_default,
                          '执行失败' + text_3)
        self.assertEquals(show_wal_level, self.wal_level_default,
                          '执行失败' + text_3)
        self.assertTrue(result_1 and result_2 and result_3, '执行失败' + text_3)
        self.logger.info(f"-----{os.path.basename(__file__)} end-----")